package git.soulbgm.listener;

import com.rabbitmq.client.Channel;
import git.soulbgm.pojo.MessageDto;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMQ侦听器
 *
 * @author SoulBGM
 * @date 2022/09/06 18:12
 */
@Component
public class RabbitmqListener {

    private final Logger log = LoggerFactory.getLogger(RabbitmqListener.class);

    private final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    /**
     * 手动创建，需在RabbitMQ中手动创建 manual 队列，否则报错
     *
     * @param dto 消息
     */
    @RabbitListener(queues = "manual")
    public void handle1(Map<String, Object> dto, Message message) {
        Date createTime = (Date) dto.get("createTime");
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 接收到 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), dto.get("messageId"), dto.get("message"),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }

    /**
     * queuesToDeclare = @Queue("test") 队列不存在时自动创建队列
     *
     * @param dto 消息
     */
    @RabbitListener(queuesToDeclare = @Queue("automatic"))
    public void handle2(MessageDto dto, Message message) {
        Date createTime = dto.getCreateTime();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 接收到 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), dto.getMessageId(), dto.getMessage(),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }

    /**
     * 自动创建队列，Exchange 与 Queue绑定
     *
     * @param message 消息
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("exchangeQueue"),
            key = "ab",
            exchange = @Exchange("exchange1")
    ))
    public void handle3(MessageDto dto, Message message) {
        Date createTime = dto.getCreateTime();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 交换机 {} 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), messageProperties.getReceivedExchange(), dto.getMessageId(), dto.getMessage(),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }

    /**
     * 自动创建队列和延迟交换机，Exchange 与 Queue绑定
     *
     * @param dto dto
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(
                    name = "exchangeQueueDelay",
                    durable = "true"),
            key = "akk",
            exchange = @Exchange(
                    name = "exchange_delay",
                    type = "x-delayed-message",
                    arguments = @Argument(name = "x-delayed-type", value = "direct"))
    ))
    public void handle4(MessageDto dto, Message message) {
        Date createTime = dto.getCreateTime();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 交换机 {} 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), messageProperties.getReceivedExchange(), dto.getMessageId(), dto.getMessage(),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }

    /*@RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(name = "exchangeQueueDeadLetter", autoDelete = "false", arguments = {
                            //绑定死信交换机
                            @Argument(name = "x-dead-letter-exchange", value = "dead_letter_exchange"),
                            //绑定死信队列
                            @Argument(name = "x-dead-letter-routing-key", value = "dead_queue"),
                            //配置最大存活时间
                            @Argument(name = "x-message-ttl", value = "10000", type = "java.lang.Integer")
                    }),
                    //配置普通交换机
                    exchange = @Exchange(value = "exchange_dead_letter"),
                    key = "ar"),
    })
    public void handle5(MessageDto dto, Message message) throws IOException {
        Date createTime = dto.getCreateTime();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 交换机 {} 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), messageProperties.getReceivedExchange(), dto.getMessageId(), dto.getMessage(),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }*/

    @Autowired
    private AmqpAdmin amqpAdmin;

    /**
     * 初始化死信队列
     */
    @PostConstruct
    public void initDeadLetterQueue() {
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", "dead_letter_exchange");
        args.put("x-dead-letter-routing-key", "dead_queue");
        args.put("x-message-ttl", 10000);

        org.springframework.amqp.core.Queue orderQueue = new org.springframework.amqp.core.Queue("order_not_paying", true, false, false, args);
        DirectExchange orderExchange = new DirectExchange("order_exchange", true, false);
        amqpAdmin.declareQueue(orderQueue);
        amqpAdmin.declareExchange(orderExchange);
        amqpAdmin.declareBinding(BindingBuilder.bind(orderQueue).to(orderExchange).with("order"));
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("QueueDeadLetter"),
            key = "dead_queue",
            exchange = @Exchange("dead_letter_exchange")
    ))
    public void handle5(MessageDto dto, Message message) {
        Date createTime = dto.getCreateTime();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 交换机 {} 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), messageProperties.getReceivedExchange(), dto.getMessageId(), dto.getMessage(),
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
    }

    /**
     * 自动创建队列，Exchange 与 Queue绑定
     *
     * @param message 消息
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("exchangeQueueTransaction"),
            key = "ap",
            exchange = @Exchange("exchange_transaction")
    ))
    public void handle6(MessageDto dto, Message message, Channel channel) throws IOException {
        Date createTime = dto.getCreateTime();
        String msg = dto.getMessage();
        MessageProperties messageProperties = message.getMessageProperties();
        log.info("队列 {} 交换机 {} 消息ID:{}, 消息内容:{}, 消息创建时间:{}", messageProperties.getConsumerQueue(), messageProperties.getReceivedExchange(), dto.getMessageId(), msg,
                LocalDateTime.ofInstant(createTime.toInstant(), ZoneId.systemDefault()).format(dtf));
        if (msg.contains("success")) {
            channel.basicAck(messageProperties.getDeliveryTag(), false);
        } else {
            channel.basicNack(messageProperties.getDeliveryTag(), false, false);
        }
    }

}
